package com.sunzm.flink.datastream.scala.time

import java.lang
import java.time.Duration

import com.alibaba.fastjson.JSON
import org.apache.commons.lang3.StringUtils
import org.apache.commons.lang3.time.{DateUtils, FastDateFormat}
import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}
import org.apache.flink.streaming.api.TimerService
import org.apache.flink.streaming.api.functions.{KeyedProcessFunction, ProcessFunction}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector
import org.slf4j.{Logger, LoggerFactory}

/**
 * 单个并行度下面的WaterMark生成示例程序
 */
object ScalaOnePartitionWartermarkDemo {
  private val logger: Logger = LoggerFactory.getLogger(this.getClass.getName.stripSuffix("$"))
  private val isLocal = true

  def main(args: Array[String]): Unit = {
    //1.创建执行的环境
    val env: StreamExecutionEnvironment = if (isLocal) {
      StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()
    } else {
      StreamExecutionEnvironment.getExecutionEnvironment
    }

    //设置全局并行度为1
    env.setParallelism(1)

    //设置周期性水印(水位线/Watermark)的间隔,默认200毫秒
    env.getConfig.setAutoWatermarkInterval(500L)

    //从socket端口获取数据
    val dataStream: DataStream[String] = env.socketTextStream("82.156.210.70", 9999)

    //WaterMark是针对事件时间而言的，处理时间语义下，没有WaterMark的概念
    //不存在乱序数据的情况下，WaterMark不需要设置最大乱序程序，与事件时间相同
    //但是查看源码，这种情况下实际上 WaterMark 比事件时间小1，所以需要一个比当前最大事件时间大1秒的事件，才能触发
    ascWaterMarkGen(dataStream)

    //5.执行
    env.execute(this.getClass.getSimpleName.stripSuffix("$"))
  }

  def ascWaterMarkGen(dataStream: DataStream[String]) = {
    //定义WatermarkStrategy
    /**
     * forMonotonousTimestamps内部调用的是AscendingTimestampsWatermarks
     * BoundedOutOfOrdernessWatermarks继承BoundedOutOfOrdernessWatermarks
     * 只不过，乱序时间设置的是0，等价于  forBoundedOutOfOrderness[String](Duration.ofSeconds(0)
     *
     * forBoundedOutOfOrderness的参数表示最大乱序程序，设置成0，就表示 Watermark 就等于事件时间
     *
     * 实际上Watermark = 最大事件时间 - 1
     *
     *  maxTimestamp = Math.max(maxTimestamp, eventTimestamp)
     *  output.emitWatermark(new Watermark(maxTimestamp - outOfOrdernessMillis - 1));
     *  outOfOrdernessMillis = 0，所以
     *
     *   Watermark = maxTimestamp - 1
     */
    //val watermarkStrategy = WatermarkStrategy.forMonotonousTimestamps[String]
    val watermarkStrategy = WatermarkStrategy.forBoundedOutOfOrderness[String](Duration.ofSeconds(0))
      //val watermarkStrategy = WatermarkStrategy.forBoundedOutOfOrderness[String](Duration.ofSeconds(5))
      .withTimestampAssigner(new SerializableTimestampAssigner[String] {
        override def extractTimestamp(element: String, recordTimestamp: Long): Long = {

          //{"callId":"call-202106260001","dateTimeStr":"20210627165616000"}
          //{"callId":"call-202106260002","dateTimeStr":"20210627165621000"}
          val jSONObject = JSON.parseObject(element)

          val dateTimeStr: String = jSONObject.getString("dateTimeStr")

          DateUtils.parseDate(dateTimeStr, "yyyyMMddHHmmssSSS").getTime
        }
      })

    //给数据流设置 时间时间提取器和watermark生成策略
    dataStream.assignTimestampsAndWatermarks(watermarkStrategy)
      .keyBy(line => {
        val jSONObject = JSON.parseObject(line)

        jSONObject.getString("callId")
      })
      //开始处理数据
      .process[String](new KeyedProcessFunction[String, String, String] {

        override def processElement(value: String,
                                    ctx: KeyedProcessFunction[String, String, String]#Context,
                                    out: Collector[String]) = {

          //事件时间
          val eventTime: Long = ctx.timestamp()

          val timerService: TimerService = ctx.timerService()

          //当前的 Watermark
          //刚启动是 Watermark 等于Long的最小值
          val currentWatermark: Long = timerService.currentWatermark()

          //以事件时间为参考点，创建一个5秒后执行的定时器
          val timerMillis: Long = eventTime + Duration.ofSeconds(5).toMillis

          val fastDateFormat: FastDateFormat = FastDateFormat.getInstance("yyyyMMddHHmmssSSS")

          val eventTimeStr: String = fastDateFormat.format(eventTime)
          val currentWatermarkStr: String = if(currentWatermark < 0) "初始值" else fastDateFormat.format(currentWatermark)
          val timerMillisStr: String = fastDateFormat.format(timerMillis)

          """
            |
            |""".stripMargin
          logger.info(
            s""" 收到消息: ${value},
                 |当前事件时间: ${eventTimeStr},
                 |当前Watermark时间: ${currentWatermarkStr},
                 |注册的定时器时间: ${timerMillisStr},
               """.stripMargin)

          timerService.registerEventTimeTimer(timerMillis)

        }

        override def onTimer(timestamp: Long,
                             ctx: KeyedProcessFunction[String, String, String]#OnTimerContext,
                             out: Collector[String]) = {

          val fastDateFormat: FastDateFormat = FastDateFormat.getInstance("yyyyMMddHHmmssSSS")

          val timestampStr: String = fastDateFormat.format(timestamp)

          val currentWatermark: Long = ctx.timerService().currentWatermark()
          val currentWatermarkStr: String = fastDateFormat.format(currentWatermark)

          logger.info(s"定时器触发: ${timestampStr}, 当前Watermark: ${currentWatermarkStr}")

          out.collect(s"${ctx.getCurrentKey} 超时...")

        }
      }).print("outPut")
  }
}
